class Consumer {
    constructor(clientId, brokers, groupId, topic, my_dao, executor) {
        this.clientId = clientId
        this.brokers = brokers
        this.groupId = groupId
        this.topic = topic
        this.my_dao = my_dao
        this.executor = executor
    }
    async start() {
        const { Kafka } = require('kafkajs')
        // 创建一个 Kafka 实例，配置客户端ID和代理服务器地址（broker）
        const kafka = new Kafka({
            clientId: this.clientId, // 客户端ID，用于在 Kafka 中标识此客户端
            //brokers: ['localhost:9092'], // 代理服务器地址（broker），这里使用本地地址和默认端口
            brokers: this.brokers
            //createPartitioner: Partitioners.LegacyPartitioner
        })

        // 创建一个消费者实例，指定消费者组ID
        const consumer = await kafka.consumer({ groupId: this.groupId })
        // 连接到 Kafka 代理服务器
        await consumer.connect()

        // 订阅指定主题的消息，从头开始消费
        await consumer.subscribe({ topic: this.topic, fromBeginning: true })

        // 启动消费者并处理每条消息
        await consumer.run({
            eachMessage: async ({ topic, partition, message }) => {
                //检查数据库是否健康
                const isHealthy = await this.my_dao.isDatabaseHealthy();
                if (!isHealthy) {
                    console.error('数据库连接不健康，无法执行任务');
                    return;
                }

                let job = null
                try {
                    job = JSON.parse(message.value)
                    if(!job || !job.id) {
                        throw new Error("任务数据格式错误:缺少id");
                    }
                    const res=await this.executor.executeJob(job)
                    await this.my_dao.addJobRecord(res.jobId, res.data, res.status)
                } catch (error) {
                    console.log(error)
                    if (job) await this.my_dao.addJobRecord(job.id, error.toString(), "error")
                    else await this.my_dao.addErrorRecord(message.value, error)
                }


            },
        })
    }
}

module.exports = Consumer;